package com.lzy.gmall.realtime.app.dwd;

import com.lzy.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/*
*
*                   购物车信息（cart_info）与 交易域订单明细事实表 关联
*
* */
public class Dwd_Transaction_field_order_details_Pair_Table {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        1. 创建执行环境并设置并行度
        env.setParallelism(1);

//        2. 创建表环境并配置状态
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        3.当某个 key 的数据超过 5 秒没有更新时，Flink 会自动清理该 key 的状态，释放内存资源
//        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
//        4.  从Kafka读取业务数据库变更日志
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("order_pre_process_211127"));

//        5.创建表格添加字段
//        dwd_order_pre       cart_info
//        购物车信息（cart_info）
        Table cart_info = tableEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['sku_id'] sku_id, " +
                "data['cart_price'] cart_price, " +
                "data['sku_num'] sku_num, " +
                "data['img_url'] img_url, " +
                "data['sku_name'] sku_name, " +
                "data['is_checked'] is_checked, " +
                "data['create_time'] create_time, " +
                "data['operate_time'] operate_time, " +
                "data['is_ordered'] is_ordered, " +
                "data['order_time'] order_time, " +
                "data['source_type'] source_type, " +
                "data['source_id'] source_id, " +
                "`pt` " +
                "from topic_db " +
                "where `table` = 'cart_info' " +
                "and `type` = 'update'");
        tableEnv.createTemporaryView("cart_info", cart_info);
//        tableEnv.toRetractStream(cart_info, Row.class).print();

//      交易域订单明细事实表
        tableEnv.executeSql("" +
                "CREATE TABLE dwd_order_pre_test( " +
                "    `id` STRING COMMENT '订单明细ID',\n" +
                "    `order_id` STRING COMMENT '订单ID',\n" +
                "    `user_id` STRING COMMENT '用户ID',\n" +
                "    `sku_id` STRING COMMENT '商品SKU ID',\n" +
                "    `sku_name` STRING COMMENT '商品名称',\n" +
                "    `order_price` STRING COMMENT '订单价格',\n" +
                "    `sku_num` STRING COMMENT '商品数量',\n" +
                "    `create_time` STRING COMMENT '创建时间',\n" +
                "    `source_type_id` STRING COMMENT '来源类型ID',\n" +
                "    `source_type_name` STRING COMMENT '来源类型名称',\n" +
                "    `source_id` STRING COMMENT '来源ID',\n" +
                "    `split_total_amount` STRING COMMENT '分摊总金额',\n" +
                "    `split_activity_amount` STRING COMMENT '分摊活动金额',\n" +
                "    `split_coupon_amount` STRING COMMENT '分摊优惠券金额',\n" +
                "    `consignee` STRING COMMENT '收货人',\n" +
                "    `consignee_tel` STRING COMMENT '收货人电话',\n" +
                "    `total_amount` STRING COMMENT '订单总金额',\n" +
                "    `order_status` STRING COMMENT '订单状态',\n" +
                "    `payment_way` STRING COMMENT '支付方式',\n" +
                "    `delivery_address` STRING COMMENT '配送地址',\n" +
                "    `order_comment` STRING COMMENT '订单评论',\n" +
                "    `trade_body` STRING COMMENT '交易内容',\n" +
                "    `operate_time` STRING COMMENT '操作时间',\n" +
                "    `expire_time` STRING COMMENT '过期时间',\n" +
                "    `process_status` STRING COMMENT '处理状态',\n" +
                "    `tracking_no` STRING COMMENT '物流单号',\n" +
                "    `province_id` STRING COMMENT '省份ID',\n" +
                "    `activity_reduce_amount` STRING COMMENT '活动减免金额',\n" +
                "    `coupon_reduce_amount` STRING COMMENT '优惠券减免金额',\n" +
                "    `original_total_amount` STRING COMMENT '原始总金额',\n" +
                "    `feight_fee` STRING COMMENT '运费',\n" +
                "    `feight_fee_reduce` STRING COMMENT '运费减免',\n" +
                "    `refundable_time` STRING COMMENT '可退款时间',\n" +
                "    `order_detail_activity_id` STRING COMMENT '订单明细活动ID',\n" +
                "    `activity_id` STRING COMMENT '活动ID',\n" +
                "    `activity_rule_id` STRING COMMENT '活动规则ID',\n" +
                "    `shop_id` BigInt COMMENT '商铺ID',\n" +// 自拟 商铺 shop_id
                "    `ts` STRING COMMENT '时间戳',\n" +
                "    `row_op_ts` TIMESTAMP_LTZ(3) COMMENT '行操作时间戳'\n" +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_order_pre", "pay_detail_suc_order_211127"));





        final Table table5 = tableEnv.sqlQuery("SELECT " +
                "   ci.id AS id, " +
                "   od.id AS order_detail_id, " +
                "   od.order_id, " +
                "   ci.user_id, " +
                "   ci.sku_id, " +
                "   COALESCE(od.sku_name, ci.sku_name) AS sku_name, " +
                "   COALESCE(od.order_price, ci.cart_price) AS order_price, " +
                "   ci.sku_num, " +
                "   od.create_time, " +
                "   od.source_type_id, " +
                "   od.source_type_name, " +
                "   od.source_id, " +
                "   od.split_total_amount, " +
                "   od.split_activity_amount, " +
                "   od.split_coupon_amount, " +
                "   od.consignee, " +
                "   od.consignee_tel, " +
                "   od.total_amount, " +
                "   od.order_status, " +
                "   od.payment_way, " +
                "   od.delivery_address, " +
                "   od.order_comment, " +
                "   od.trade_body, " +
                "   od.operate_time, " +
                "   od.expire_time, " +
                "   od.process_status, " +
                "   od.tracking_no, " +
                "   od.province_id, " +
                "   od.activity_reduce_amount, " +
                "   od.coupon_reduce_amount, " +
                "   od.original_total_amount, " +
                "   od.feight_fee, " +
                "   od.feight_fee_reduce, " +
                "   od.refundable_time, " +
                "   od.id AS order_detail_activity_id, " +
                "   od.activity_id, " +
                "   od.activity_rule_id, " +
                "   ci.is_ordered, " +
                "   CAST(CEIL(RAND() * 100) AS BigInt) AS shop_id, " +
                "   CAST(CURRENT_TIMESTAMP AS STRING) AS ts, " +
                "   CURRENT_ROW_TIMESTAMP() AS row_op_ts " +
                "FROM cart_info ci " +
                "LEFT JOIN dwd_order_pre_test od " +
                "ON ci.sku_id = od.sku_id AND ci.user_id = od.user_id ");

            tableEnv.createTemporaryView("result_table", table5);


//      创建生成包含购物车信息和订单信息的宽表数据
        tableEnv.executeSql("" +
                "CREATE TABLE dwd_order_pre ( " +
                "    id STRING COMMENT '购物车ID',\n" +
                "    order_detail_id STRING COMMENT '订单明细ID',\n" +
                "    order_id STRING COMMENT '订单ID',\n" +
                "    user_id STRING COMMENT '用户ID',\n" +
                "    sku_id STRING COMMENT '商品SKU ID',\n" +
                "    sku_name STRING COMMENT '商品名称',\n" +
                "    order_price STRING COMMENT '订单价格',\n" +
                "    sku_num STRING COMMENT '商品数量',\n" +
                "    create_time STRING COMMENT '创建时间',\n" +
                "    source_type_id STRING COMMENT '来源类型ID',\n" +
                "    source_type_name STRING COMMENT '来源类型名称',\n" +
                "    source_id STRING COMMENT '来源ID',\n" +
                "    split_total_amount STRING COMMENT '分摊总金额',\n" +
                "    split_activity_amount STRING COMMENT '分摊活动金额',\n" +
                "    split_coupon_amount STRING COMMENT '分摊优惠券金额',\n" +
                "    consignee STRING COMMENT '收货人',\n" +
                "    consignee_tel STRING COMMENT '收货人电话',\n" +
                "    total_amount STRING COMMENT '订单总金额',\n" +
                "    order_status STRING COMMENT '订单状态',\n" +
                "    payment_way STRING COMMENT '支付方式',\n" +
                "    delivery_address STRING COMMENT '配送地址',\n" +
                "    order_comment STRING COMMENT '订单评论',\n" +
                "    trade_body STRING COMMENT '交易内容',\n" +
                "    operate_time STRING COMMENT '操作时间',\n" +
                "    expire_time STRING COMMENT '过期时间',\n" +
                "    process_status STRING COMMENT '处理状态',\n" +
                "    tracking_no STRING COMMENT '物流单号',\n" +
                "    province_id STRING COMMENT '省份ID',\n" +
                "    activity_reduce_amount STRING COMMENT '活动减免金额',\n" +
                "    coupon_reduce_amount STRING COMMENT '优惠券减免金额',\n" +
                "    original_total_amount STRING COMMENT '原始总金额',\n" +
                "    feight_fee STRING COMMENT '运费',\n" +
                "    feight_fee_reduce STRING COMMENT '运费减免',\n" +
                "    refundable_time STRING COMMENT '可退款时间',\n" +
                "    order_detail_activity_id STRING COMMENT '订单明细活动ID',\n" +
                "    activity_id STRING COMMENT '活动ID',\n" +
                "    activity_rule_id STRING COMMENT '活动规则ID',\n" +
                "    is_ordered STRING COMMENT '是否已下单',\n" +
                "    shop_id BigInt COMMENT '商铺ID',\n" +
                "    ts STRING COMMENT '时间戳',\n" +
                "    row_op_ts TIMESTAMP_LTZ(3) COMMENT '行操作时间戳',\n" +
                "    PRIMARY KEY (id) NOT ENFORCED" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_order_pre"));

        tableEnv.executeSql("insert into dwd_order_pre select * from result_table");
        tableEnv.executeSql("select * from dwd_order_pre").print();

        env.execute();
    }
}